package com.qyer.commons.concurrent.assemblingline;

import static com.qyer.commons.concurrent.assemblingline.AssemblingLineStatus.INITED;
import static com.qyer.commons.concurrent.assemblingline.AssemblingLineStatus.STANDBY;
import static com.qyer.commons.concurrent.assemblingline.ProcessMode.BULK;
import static com.qyer.commons.utils.CommonUtils.timeCost;

import com.qyer.commons.CheckStatus;
import com.qyer.commons.concurrent.ArrayListBulk;
import com.qyer.commons.concurrent.GenericLinkedBlockingQueue;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * User: Z J Wu Date: 2016/06/03 Time: 10:05 Package: com.qyer.commons.concurrent
 */
public class BulkAssembler<T> implements Runnable, AssembleBulkConveyors<T>,
  AssembleBulkNotifier<T>, CheckStatus {

  private static final Logger LOGGER = LoggerFactory.getLogger(BulkAssembler.class);

  protected String name;
  protected String groupName;
  protected GenericLinkedBlockingQueue<ArrayListBulk<T>> in;
  protected GenericLinkedBlockingQueue<ArrayListBulk<T>> out;
  protected Notifier<ArrayListBulk<T>> notifier;
  protected AbstractHandler<ArrayListBulk<T>, ArrayListBulk<T>> handler;
  protected Differential<T> differential;
  protected StatisticContext context;
  protected AssemblingLineStatus status;
  protected int bulkSize;

  public BulkAssembler(String groupName, String name,
                       AbstractHandler<ArrayListBulk<T>, ArrayListBulk<T>> handler,
                       StatisticContext context) {
    this.groupName = groupName;
    this.name = name;
    this.handler = handler;
    this.context = context;
    this.status = check();
  }

  @Override
  public void run() {
    LOGGER.info("Assembler started.");
    try {
      while (true) {
        ArrayListBulk<T> bulk = in.take();
        if (bulk.isPill()) {
          break;
        }
        if (bulk.isEmpty()) {
          continue;
        }
        long t1 = System.currentTimeMillis();
        try {
          bulk = handler.handle(bulk);
          context.reportSuccess(groupName, BULK);
        } catch (HandleException e) {
          LOGGER.warn("Handler cannot handle this bulk.", e);
          context.reportFailure(groupName, BULK);
          continue;
        } finally {
          context.reportTimeCost(groupName, timeCost(t1), BULK);
        }
        put2Conveyor(bulk);
      }
      // 最后要清空差速器里的剩余内容
      if (differential != null) {
        differential.flush();
      }
      LOGGER.info("Assembler stopped.");
    } catch (InterruptedException e) {
      LOGGER.error("Assembler is interrupted", e);
    } catch (Exception e) {
      LOGGER.error("Assembler aborted with error.", e);
    } finally {
      notifier.finishAndNotify();
    }
  }

  public void put2Conveyor(ArrayListBulk<T> bulk) throws AssemblingLineException,
    InterruptedException {
    if (differential == null) {
      out.put(bulk);
    } else {
      differential.addCheckPush(bulk);
    }
  }

  public String getName() {
    return name;
  }

  @Override
  public void assembleConveyors(GenericLinkedBlockingQueue<ArrayListBulk<T>>... conveyors) {
    if (ArrayUtils.isNotEmpty(conveyors) && conveyors.length > 1) {
      this.in = conveyors[0];
      this.out = conveyors[1];
      this.status = check();
    }
  }

  public void assembleNotifier(Notifier<ArrayListBulk<T>> notifier) {
    this.notifier = notifier;
    this.status = check();
  }

  public void setDifferential(Differential<T> differential) {
    this.differential = differential;
    this.status = check();
  }

  public int getBulkSize() {
    return bulkSize;
  }

  public void setBulkSize(int bulkSize) {
    this.bulkSize = bulkSize;
  }

  @Override
  public AssemblingLineStatus check() {
    if (bulkSize <= 0 || StringUtils.isEmpty(
      name) || in == null || out == null || notifier == null || handler == null || differential
      == null || context == null) {
      return INITED;
    }
    return STANDBY;
  }
}
